package day09

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
/**
  * 一个简单的Streaming的wordcount
  * 数据从netcat中获取
  * 注意点：streaming的代码中local必须大于等于2，一个用于driver,一个用于excutor，sc不能stop，那些算子是处理DStream而不是RDD
  */
object streamingDemo {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setAppName("streamingDemo").setMaster("local[*]")
    val sc = new SparkContext(conf)
    // 创建Streaming的上下文，第二个参数数Stream多长时间获取一次数据
    // Durations.seconds(5) == Seconds(5)
    val ssc = new StreamingContext(sc, Durations.seconds(5))//批次间隔可以设置成毫秒：Milliseconds(2000)

    // 通过SocketTextStream获取netcat的数据，这种获取方式stream会把获取到的数据以缓存的方式放到指定缓存级别的地方
    // 前提是已经在netcat 输入nc -l -p 8888开启这个端口的网络服务。不加 -p 的话会报连接不上错误。并且nc就没有-k的参数，或者说你看你开了几个代码在运行，默认是只能一个区接受nc的消息。
    val dStream: ReceiverInputDStream[String] = ssc.socketTextStream("master",8888, StorageLevel.MEMORY_AND_DISK)
    // 分析数据
    val res: DStream[(String, Int)] = dStream.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    //打印到控制台
    res.print()
    //提交任务到集群
    ssc.start()
    //线程等待
    ssc.awaitTermination()



  }
}
/*
用transform可以操作DStream里面的rdd，返回的还是DStream，既然DStream本身就可以使用自己的Transformations和Output OPtions得到数据分析结果，为什么还用transform来操作RDD得到数据分析结果呢？这是一位DStream自己本身的Transformations和output options的算子太少了，没有RDD的功能强大，因此用RDD的算子。
val res: DStream[(String, Int)] = dstream.transform(rdd => {
	rdd.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)	//RDD的算子
})
*/